package com.zhangsr.zus.rpc.zus;

import com.zhangsr.zus.common.Constants;
import com.zhangsr.zus.common.URL;
import com.zhangsr.zus.remoting.api.Channel;
import com.zhangsr.zus.remoting.api.exception.RemotingException;
import com.zhangsr.zus.remoting.api.exchange.*;
import com.zhangsr.zus.remoting.api.exchange.support.ExchangeHandlerAdapter;
import com.zhangsr.zus.rpc.api.Exporter;
import com.zhangsr.zus.rpc.api.Invocation;
import com.zhangsr.zus.rpc.api.Invoker;
import com.zhangsr.zus.rpc.api.exception.RpcException;
import com.zhangsr.zus.rpc.api.protocol.AbstractProtocol;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author: zhangsr
 */
public class ZusRpcProtocol extends AbstractProtocol {

    private static Logger logger = Logger.getLogger(ZusRpcProtocol.class);

    public static final String NAME = "zus";

    public static final int DEFAULT_PORT = 20880;

    private final Map<String, ExchangeServer> serverMap = new ConcurrentHashMap<String, ExchangeServer>();

    private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
        // 接收Consumer请求，反射方法类
        public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                Invocation inv = (Invocation) message;
                Invoker<?> invoker = getInvoker(channel, inv);
                return invoker.invoke(inv);
            }
            throw new RemotingException(channel, "Unsupported request: " + message == null ? null : (message.getClass().getName() + ": " + message) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
        }

        public void received(Channel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                // 从NettyHandler接收Message
                reply((ExchangeChannel) channel, message);
            } else {
                super.received(channel, message);
            }
        }
    };

    Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
        int port = channel.getLocalAddress().getPort();
        String serviceKey = serviceKey(port, inv.getAttachments().get(Constants.VERSION_KEY), inv.getAttachments().get(Constants.GROUP_KEY));

        ZusRpcExporter<?> exporter = (ZusRpcExporter<?>) exporterMap.get(serviceKey);
        if (exporter == null) {
            throw new RemotingException(channel, "Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch " + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + inv);
        }
        return exporter.getInvoker();
    }

    public int getDefaultPort() {
        return DEFAULT_PORT;
    }

    public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();

        // export service.
        String key = serviceKey(url);
        ZusRpcExporter<T> exporter = new ZusRpcExporter<T>(invoker, key, exporterMap);
        exporterMap.put(key, exporter);

        //export an service
        openServer(url);

        return exporter;
    }

    private void openServer(URL url) {
        // find server.
        String key = url.getAddress();
        //client 也可以暴露一个只有server可以调用的服务。
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
        if (isServer) {
            ExchangeServer server = serverMap.get(key);
            if (server == null) {
                serverMap.put(key, createServer(url));
            } else {
            }
        }
    }

    private ExchangeServer createServer(URL url) {
        ExchangeServer server;
        try {
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }
        return server;
    }

    public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
        // create rpc invoker.
        ZusRpcInvoker<T> invoker = new ZusRpcInvoker<T>(serviceType, url, getClients(url), invokers);
        invokers.add(invoker);
        return invoker;
    }

    private ExchangeClient[] getClients(URL url) {
        int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
        //如果connections不配置，则共享连接，否则每服务每连接
        if (connections == 0) {
            connections = 1;
        }
        ExchangeClient[] clients = new ExchangeClient[connections];
        for (int i = 0; i < clients.length; i++) {
            clients[i] = initClient(url);
        }
        return clients;
    }

    /**
     * 创建新连接.
     */
    private ExchangeClient initClient(URL url) {
        ExchangeClient client = null;
        try {
            client = Exchangers.connect(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to create remoting client for service(" + url
                    + "): " + e.getMessage(), e);
        }
        return client;
    }

    public void destroy() {

    }
}
